package com.haizhi.sparksql.demo;

import com.haizhi.sparksql.bean.User;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.*;

public class ReadCSV {
    public static void main(String[] args) {
        //1. 创建配置对象
        SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]");

        //2. 获取sparkSession
        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();

        //3. 编写代码
        DataFrameReader reader = spark.read();

        // 添加参数  读取csv
        Dataset<Row> userDS = reader
                .option("header", "true")//默认为false 不读取列名
                .option("sep",",") // 默认为, 列的分割
                // 不需要写压缩格式  自适应
                .csv("sparkSQL/input/user.csv");

        userDS.show();

        // 转换为user的ds
        // 直接转换类型会报错  csv读取的数据都是string
//        Dataset<User> userDS1 = userDS.as(Encoders.bean(User.class));
        userDS.printSchema();

        Dataset<User> userDS1 = userDS.map(new MapFunction<Row, User>() {
            @Override
            public User call(Row value) throws Exception {
                return new User(Long.valueOf(value.getString(0)), value.getString(1));
            }
        }, Encoders.bean(User.class));

        userDS1.show();

        // 写出为csv文件
        DataFrameWriter<User> writer = userDS1.write();

        writer.option("seq",";")
                .option("header","true")
//                .option("compression","gzip")// 压缩格式
                // 写出模式
                // append 追加
                // Ignore 忽略本次写出
                // Overwrite 覆盖写
                // ErrorIfExists 如果存在报错
                .mode(SaveMode.Append)
                .csv("sparkSQL/output");

        //4. 关闭sparkSession
        spark.close();
    }
}
